package com.kafka.demo.controller;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.kafka.demo.cons.KafkaTopic;
import com.kafka.demo.domain.req.MsgReq;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.time.LocalDateTime;
import java.util.UUID;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

@RestController
@RequestMapping(value ="/demo")
@Slf4j
public class ProducerController {

    //kafka-producers.service 做了自动拼接的bean注册命名，方便多个生产者区分
    @Autowired
    @Qualifier("myDemoKafkaTemplate")
    private KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    private ObjectMapper objectMapper;

    /**
     * 普通发送
     * @param req
     * @return
     * @throws JsonProcessingException
     */
    @RequestMapping(value ="/cluster-send")
    public String sendMsg(@RequestBody MsgReq req) throws JsonProcessingException {
        req.setUuid(UUID.randomUUID().toString());
        String json = objectMapper.writeValueAsString(req);
        try {
            //同步发送, 可以在使用过程中将broker 宕机，检验是否同步还是异步
            SendResult<String, String> result = kafkaTemplate.send(KafkaTopic.MY_TOPIC, json).get();
            // 成功发送
            log.info("✅ 消息发送成功，分区：{}, 偏移量：{} " ,result.getRecordMetadata().partition(),   result.getRecordMetadata().offset());
        } catch (ExecutionException | InterruptedException e) {
            log.error(e.getMessage(),e);
            return "发送异常";
        }
        return "发送成功";
    }

    /**
     * 广播发送
     * @param req
     * @return
     * @throws JsonProcessingException
     */
    @RequestMapping(value ="/broadcast-send")
    public String broadcastMsg(@RequestBody MsgReq req) throws JsonProcessingException {
        req.setUuid(UUID.randomUUID().toString());
        String json = objectMapper.writeValueAsString(req);
        try {
            //同步发送
            SendResult<String, String> result = kafkaTemplate.send(KafkaTopic.MY_TOPIC2, json).get();
            // 成功发送
            log.info("✅ 消息发送成功，分区：{}, 偏移量：{} " ,result.getRecordMetadata().partition(),   result.getRecordMetadata().offset());
        } catch (ExecutionException | InterruptedException e) {
            log.error(e.getMessage(),e);
            return "发送异常";
        }
        return "发送成功";
    }




}
